package com.mchat.common.service;

import com.mchat.common.dao.FileHashDAO;
import com.mchat.common.domain.FileHash;
import com.mchat.constant.RedisConstant;
import com.mchat.exceptions.CommonException;
import com.mchat.util.JsonUtil;
import com.mchat.util.MailUtil;
import com.mchat.util.SnowflakeIdGenerator;
import io.minio.MinioAsyncClient;
import io.minio.PostPolicy;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;

import java.time.Duration;
import java.time.ZonedDateTime;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
@Slf4j
@RequiredArgsConstructor
public class CommonService {


    private final ReactiveStringRedisTemplate redisTemplate;

    private final MailUtil mailUtil;

    private final MinioAsyncClient minioAsyncClient;

    private final SnowflakeIdGenerator snowflakeIdGenerator;

    private final FileHashDAO fileHashDAO;
    private final ReceiverOptions<Long, String> receiverOptions;
    private final JsonUtil jsonUtil;

    @SneakyThrows
    public Map<String, String> getPreSignedURL(Long uid) {
        String fileName = snowflakeIdGenerator.nextId() +".png";
        PostPolicy policy = new PostPolicy("profile-img", ZonedDateTime.now().plusMinutes(30));

        policy.addEqualsCondition("key", fileName);
        policy.addContentLengthRangeCondition(1, 2 * 1024 * 1024);
        Map<String, String> data = minioAsyncClient.getPresignedPostFormData(policy);
        data.put("key",fileName);
        return data;
    }

    public Mono<Void> sendCode(String receiver) {
        // 生成一个 6 位的随机验证码
        int captcha = (int) (Math.random() * 1000000);
        // 使用 String.format 来确保验证码是 6 位，首位补零
        String code = String.format("%06d", captcha);
        mailUtil.sendTextVerifyCode(receiver, code);
        return redisTemplate.opsForValue()
                .set(RedisConstant.VERIFY_CODE_FOR_REGISTER + ":" + receiver, code, Duration.ofMinutes(mailUtil.getValidTime()))
                .flatMap(b -> {
                    if (b) return Mono.empty();
                    return Mono.error(new CommonException("发送失败", 500));
                });
    }

    @PostConstruct
    public void receiveHash() {
        KafkaReceiver.create(receiverOptions.subscription(List.of("file-hash")))
                .receive()
                .bufferTimeout(100, Duration.ofSeconds(1))
                .filter(batch -> !batch.isEmpty()) // 避免空批次
                .concatMap(list -> {


                    log.info("收到{}条数据", list.size());
                    List<FileHash> collect = list.stream().map(v -> jsonUtil.toObj(v.value(), FileHash.class)).collect(Collectors.toList());
                    return fileHashDAO.batchSave(collect)
                            .flatMap(count -> list.get(list.size() - 1).receiverOffset().commit().thenReturn(count));
                }).subscribe(count -> {
                    log.info("保存了{}条数据", count);
                });

    }
}
